package study.nosql.redis.consumer;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.thread.ThreadUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.BoundStreamOperations;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * Stream 实现队列
 */
@Slf4j
@Component
public class StreamQueConsumer implements CommandLineRunner {

    @Autowired
    private BoundStreamOperations<String, String, Object> blockQueue;


    @Override
    public void run(String... args) throws Exception {
//        //
//        Thread t = ThreadUtil.newThread(() -> {
//            consumer();
//        }, "Queue-Stream-Test-Thread");
//        t.start();
    }

    public void consumer() {
        // 读参数
        StreamReadOptions options = StreamReadOptions.empty()
                .block(Duration.ofMillis(5 * 1000))
                .autoAcknowledge()
                .count(10);
        // 读位移
//        ReadOffset.latest();
//        ReadOffset.from("");
        ReadOffset offset = ReadOffset.lastConsumed();
        // 消费者
        Consumer consumer = Consumer.from("my_group", "my_name");

        while (true) {
            List<MapRecord<String, String, Object>> tt = blockQueue.read(consumer, options, offset);
            if (CollectionUtil.isEmpty(tt)) {
                log.warn(">>>>>> 队列为空");
                continue;
            }
            tt.forEach(e -> {
                RecordId recId = e.getId();
                Map<String, Object> message = e.getValue();
                log.info(">>>>>>>>> {}", message);
            });
        }
    }
}
